
import java.sql.{Connection, DriverManager, PreparedStatement}

import MySqlForeachWriter.{password, url, username}
import org.apache.spark.sql.ForeachWriter

object MySqlForeachWriter{

  private val url="jdbc:mysql://linux123:3306/mysql?characterEncoding=UTF-8&useSSL=false"  //自己的ip地址:端口号/库名
  private val username="root"   //用户名
  private val password="12345678"     //密码

}

class MySqlForeachWriter extends ForeachWriter[BusInfo]{
  var mysql:Connection =_
  //开启链接
  override def open(partitionId: Long, epochId: Long): Boolean = {
    mysql=DriverManager.getConnection(url,username,password)
    true;
  }
  //处理数据
  override def process(value: BusInfo): Unit = {
    val preparedStatement: PreparedStatement = mysql.prepareStatement("insert into bus_info (plateNumArr,lng,lat) values(?,?,?)")
    preparedStatement.setString(1,value.plateNum)
    val arr: Array[String] = value.lglat.split("_")
    preparedStatement.setString(2,arr(0))
    preparedStatement.setString(3,arr(1))
    preparedStatement.executeUpdate()
  }
  //关闭连接
  override def close(errorOrNull: Throwable): Unit = {
    if(mysql!=null) {
      mysql.close()
    }
  }
}
